package edu.kit.ksri.db.mturk.workerpools;

import java.io.Serializable;
import java.util.Calendar;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;

import com.amazonaws.mturk.requester.Assignment;

import edu.kit.ksri.db.mturk.simulations.SimMTurk;
import edu.kit.ksri.db.mturk.var.StaticInformation;
import edu.kit.ksri.db.mturk.workerpools.AbstractWorker;
import edu.kit.ksri.db.mturk.workerpools.CSP1Capable;
/**
 * Copyright (c) 2010-2011 Karlsruhe Institute of Technology (KIT), Germany
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense,
 * and/or sell copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 * 
 * --------------------------------------------------------------------------
 * 
 * Software: CSP/WMV tool for dynamic quality management of microtasks
 * http://www.ksri.kit.edu/Upload/Users/PeopleClouds/HCOMP2011_An_extendable_toolkit_for_managing_quality_of_human-based_eServices.pdf
 * 
 * Organization: Karlsruhe Institute of Technology (KIT), Germany
 * http://www.kit.edu
 * 
 * Project: PeopleClouds project of the Karlsruhe Service Research Institute(KSRI)
 * http://www.ksri.kit.edu/Default.aspx?PageId=818
 * 
 * Contributing authors: David Bermbach, Sandra Rath, Pascal Wichmann, Robert Kern
 * 
 * ---------------------------------------------------------------------------
 * 
 * Purpose: Simulates a worker
 */
public class SimWorker extends AbstractWorker implements Serializable, CSP1Capable, Runnable {

	private static final long serialVersionUID = -4711845984396767901L;

	private static final Logger basiclog = Logger.getLogger("basic." + SimWorker.class);
	
	/** the worker id */
	//private final String workerId;
	
	/** determines whether the next assignment needs to be inspected due to CSP1	 */
	private boolean mustBeInspectedForCSP1 = true;

	/**
	 * specifies how many units have been inspected in a row and have been found
	 * to be correct. If this value exceeds {@link StaticInformation}.i only
	 * fractions are inspected.
	 */
	private int inspectedCorrectSamplesInARow = 0;
	
	/**
	 * the total number of assignments of this worker which were found to be
	 * correct based on inspection
	 */
	private int totalInspectedCorrect = 0;
	
	/**
	 * the total number of assignments of this worker which were found to be
	 * incorrect based on inspection
	 */
	private int totalInspectedFalse = 0;
	
	/** number of Assignments the SimWorker as accomplished without being inspected	 */
	private int frontRunning = 0;
	
	/**
	 * stores all possible answers of this worker
	 */
	private final ConcurrentHashMap<String, Assignment> workerAssignments = new ConcurrentHashMap<String, Assignment>();

	/**
	 * the qualification value determines whether a worker may work on a given
	 * task or not. cf. also {@link StaticInformation}
	 */
	private int qualificationValue = 0;

	/** determines what the worker would do next */
	private SimWorkerState state = SimWorkerState.WaitingForNextHIT;

	
	/**
	 * references a position in possibleAnswers, is retrieved from
	 * {@link PerfSimMTurk}.
	 */
	private String currentHitId = null;
	
	/**
	 * @param workerId
	 * @param possibleAnswers
	 *            all possible answers of this worker, items may be null if the
	 *            worker cannot provide an answer to that question id
	 * @param qualificationValue
	 *            the qualification values, determines whether a worker may work
	 *            on a given task or not. cf. also {@link StaticInformation}.
	 */
	public SimWorker(String workerId, ConcurrentHashMap<String, Assignment> workerAssignments,
			int qualificationValue) {
		super();
		this.workerId = workerId;
		this.workerAssignments.putAll(workerAssignments);
		this.qualificationValue = qualificationValue;
		basiclog.info("Created worker " + workerId + " with qualification value "
				+ qualificationValue + " and " + workerAssignments.size()
				+ " answers.");
		this.totalAssignments = 10;
		this.totalInspectedCorrect = 9;
		this.totalInspectedFalse = 1;
		this.correctnessLevel = 100	* this.totalInspectedCorrect / (this.totalInspectedCorrect + this.totalInspectedFalse);
	}
	
	public SimWorker(String workerId, int totalAssignments,
			int totalInspectedCorrect, int totalInspectedFalse) {
		super();
		this.workerId = workerId;
		this.totalAssignments = totalAssignments;
		this.totalInspectedCorrect = totalInspectedCorrect;
		this.totalInspectedFalse = totalInspectedFalse;
		this.correctnessLevel = 100	* this.totalInspectedCorrect / (this.totalInspectedCorrect + this.totalInspectedFalse);
	}
	
	/**
	 * @param workerId
	 */
	public SimWorker(String workerId) {
		super();
		this.workerId = workerId;
		this.totalAssignments = 10;
		this.totalInspectedCorrect = 9;
		this.totalInspectedFalse = 1;
		this.correctnessLevel = 100	* this.totalInspectedCorrect / (this.totalInspectedCorrect + this.totalInspectedFalse);
	}

	public void run() {
		basiclog.info("Worker " + workerId + " started.");
		int timeOutCounter = 0;
		//while (SimMTurk.isRunning()) {
		while (timeOutCounter < 2) {
			switch (state) {
			case WaitingForNewQualificationValue: {
				int newQuali = 0;
				try {
					newQuali = SimMTurk.getQualificationUpdate(workerId);
					basiclog
							.info(workerId+": Got new qualification value "
									+ newQuali
									+ ", is now "
									+ (newQuali < StaticInformation.activePoolQualificationValue ? "blocked"
											: "allowed"));
				} catch (Exception e) {
					break;
				}
				this.qualificationValue = newQuali;
				if (newQuali < StaticInformation.activePoolQualificationValue) {
					// retry later...do nothing
				} else {
					frontRunning = 0;
					state = SimWorkerState.WaitingForNextHIT;
				}
				break;
			}
			case WaitingForNextHIT: {
				if (Math.random() < StaticInformation.hitAcceptanceProbabilityForSimWorkers) {
					// check whether there is a qualification update
					int newQuali = 0;
					try {
						newQuali = SimMTurk.getQualificationUpdate(workerId);
						this.qualificationValue = newQuali;
						frontRunning = 0;
						basiclog
						.info(workerId+": Got new qualification value "
								+ qualificationValue
								+ ", is now "
								+ (this.qualificationValue < StaticInformation.activePoolQualificationValue ? "blocked"
										: "allowed"));
					} catch (Exception e) {
						// no need to do anything at all
						//basiclog.debug("Worker " + workerId + " got no new qualification value.");
					}

					if (this.qualificationValue < StaticInformation.activePoolQualificationValue) {
						state = SimWorkerState.WaitingForNewQualificationValue;
						break;
					}
					// qualification is ok, now poll for a hit
					String hitID = null;
					try {
						hitID = SimMTurk.getNextHIT(this);
						if (workerAssignments.get(hitID) == null) {
						//if (!hasAssignments(hitID)){
							// return hit
							SimMTurk.addHIT(hitID);
						} else {
							state = SimWorkerState.Working;
							currentHitId = hitID;
						}
						
					}catch (NullPointerException npe) {
						//npe.printStackTrace();
						break;
					} 
					catch (Exception e) {
						//e.printStackTrace();
						break;
					}
					
				}
				break;
			}
			case Working: {
				if (currentHitId != null) {
					//Assignment assignment = workerAssignments.get(currentHitId);
					Assignment assignment = workerAssignments.remove(currentHitId);
					
					if (assignment == null) {
						// should never happen, was already checked in state
						// WaitingForNextHIT
						state = SimWorkerState.WaitingForNextHIT;
						SimMTurk.addHIT(currentHitId);
					} else {
						basiclog.info(workerId+": Working on hit " + currentHitId);
						// wait till we can return the answer
						long delay = StaticInformation.maxTaskDurationForSimWorkers
								- StaticInformation.minTaskDurationForSimWorkers;
						delay = (long) (Math.random() * (delay + 1));
						delay += StaticInformation.minTaskDurationForSimWorkers;
						try {
							Thread.sleep(delay);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						
						assignment.setSubmitTime(Calendar.getInstance());
						SimMTurk.addAnswer(currentHitId, assignment);
						currentHitId = null;
						
						
						frontRunning++;
						basiclog.debug("Frontrunning for worker " + workerId + " is currently: " + frontRunning);
						if (frontRunning < StaticInformation.frontRunningParamSimWorker)
							state = SimWorkerState.WaitingForNextHIT;
						else{
							SimMTurk.addQualificationUpdate(this.workerId, new Integer(0));
							state = SimWorkerState.WaitingForNewQualificationValue;
							basiclog.debug("Frontrunning param reached for SimWorker " + this.workerId + ", Worker is blocked");
						}
							
					}
				} else
					state = SimWorkerState.WaitingForNextHIT;
				break;
			}
			default: {
				throw new RuntimeException("Unsupported state: " + state);
			}
			}
			try {
				Thread.sleep((long)(Math.random()*StaticInformation.updateIntervalForSimWorkers));
				
				if (workerAssignments.isEmpty())  //wait another update intervall
					timeOutCounter++;
				else
					timeOutCounter = 0;
					
			} catch (InterruptedException e) {
				// does not matter because it should not happen
				e.printStackTrace();
			}

		}

	}
	
	/**
	 * Adds assignments to the worker's assignment pool
	 * @param hitID HIT to which the worker was contributing
	 * @param assignment Assignment the worker was working at
	 */
	public void addAssignment (String hitID, Assignment assignment){
		if (!workerAssignments.containsKey(hitID)){
			workerAssignments.put(hitID, assignment);
			basiclog.debug("Added assignment " + assignment.getAssignmentId() + " for HIT " + hitID + " to worker " + this.workerId);
		}else if (workerAssignments.get(hitID).equals(assignment)){
			//Assignment already in workerAssignments: Do nothing
		}else{
			basiclog.error("Caution! Assignment " + workerAssignments.get(hitID).getAssignmentId() + " for HIT " + hitID + " is being replaced by Assignment " + assignment.getAssignmentId());
		}
	}
	
	@Override
	public boolean equals(Object arg) {
		// System.out.println("EQUALS invoked for " + workerId + " and "
		// + ((Worker) arg).workerId + ", returning: "
		// + (this.workerId.equals(((Worker) arg).workerId)));
		if (!(arg instanceof SimWorker))
			return false;
		return this.workerId.equals(((SimWorker) arg).workerId);
	}
	
	public synchronized String[] toStringArray() {
		String[] res = new String[3];
		res[0] = "" + workerId;
		res[1] = "" + correctnessLevel;
		res[2] = "" + mustBeInspectedForCSP1;
		return res;
	}
	
	public int compareTo(AbstractWorker other) {
		if (!(other instanceof SimWorker))
			try {
				throw new Exception("Cannot compare workers of different types.");
			} catch (Exception e) {
				e.printStackTrace();
			}
		if (this.correctnessLevel < other.correctnessLevel)
			return 1;
		if (this.correctnessLevel > other.correctnessLevel)
			return -1;
		return 0;
	}
	
	/**
	 * updates this worker's statistics based on CSP-1. based on the parameters
	 * it updates all counter fields, determines whether the inspection mode
	 * needs to change and finally updates the field mustBeInspectedForCSP1.
	 * 
	 * @param wasInspected
	 *            whether the assignment was inspected or not
	 * @param wasCorrect
	 *            whether the inspection resulted in pass (true) or fail
	 *            (false). if wasInspected == false this parameter is ignored.
	 */
	public void addAssignmentForBinomialCSP1(boolean wasInspected,
			boolean wasCorrect) {
		
		basiclog.debug("This is worker " + this.workerId + " being updated");
		synchronized (syncLockField) {
			this.totalAssignments++;
			if (wasInspected) {
				// If the worker was not inspected, nothing needs to be done!
				if (wasCorrect) {
					this.totalInspectedCorrect++;
					this.inspectedCorrectSamplesInARow++;
				} else {
					this.totalInspectedFalse++;
					this.inspectedCorrectSamplesInARow = 0;
				}
				
				
			}
			// Update correctness level
			this.correctnessLevel = 100
				// Only inspected samples may be taken into account!
				* this.totalInspectedCorrect / (this.totalInspectedCorrect + this.totalInspectedFalse);


			// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
			// April 9, 2011: temporary fix for phi=1 problem: avoid that phi is ever getting
			// exactly 1
			// this.correctnessLevel = this.correctnessLevel * 0.99999999;
			// July 5, 2011: removed temporary fix. Phi = 1 case should be avoided by preventing
			// 100% correct qualification test results in QualificationManager.java! 
			// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

			// Update mustBeInspectedForCSP1 field
			if ( this.inspectedCorrectSamplesInARow >= StaticInformation.scenario.getI() ) {
				// If the number i of inspected correct assignments is met or exceeded
				// decide randomly whether to inspect or not.
				// (Math.random() provides random numbers between 0 and 1)
				this.mustBeInspectedForCSP1 =
					(Math.random() <= StaticInformation.scenario.getF());
			} else {
				// In i is not met, the following assignment has to be inspected
				this.mustBeInspectedForCSP1 = true;
			}
			basiclog.info("Worker " + this.workerId + " has returned " +
					this.inspectedCorrectSamplesInARow + " correct results in a row."); 
			basiclog.info("Inspection of worker required? " + this.mustBeInspectedForCSP1);
			
			if (this.correctnessLevel >= StaticInformation.minimumCorrectnessLevel){
				SimMTurk.addQualificationUpdate(this.workerId, new Integer(1));
				basiclog.debug("Worker " + workerId + " is allowed to work again.");
			} else
				basiclog.debug("Worker's correctness level is too low to continue further work: " + correctnessLevel);
		}
	}
	
	/**
	 * @return whether this worker's next assignment must be inspected or not
	 */
	public boolean isMustBeInspectedForCSP1() {
		return this.mustBeInspectedForCSP1;
	}
	
	/**
	 * @return the totalInspectedCorrect
	 */
	public int getTotalInspectedCorrect() {
		return this.totalInspectedCorrect;
	}
	
	/**
	 * @return the totalInspectedFalse
	 */
	public int getTotalInspectedFalse() {
		return this.totalInspectedFalse;
	}
	
	public boolean hasAssignments(String hitID){
		return (workerAssignments.containsKey(hitID) && workerAssignments.get(hitID) != null);
	}

}
